package com.atguigu.day09;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Example10 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        var streamTableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build());

        streamTableEnv
                .executeSql(
                        "CREATE TABLE clicks (" +
                                " `username` STRING," +
                                " `url` STRING," +
                                " `ts` TIMESTAMP(3)," +
                                " WATERMARK FOR `ts` AS `ts` - INTERVAL '3' SECONDS)" +
                                " WITH (" +
                                "   'connector' = 'filesystem'," +
                                "   'path' = '/home/yuantuzhi/flinktutorial0905/src/main/resources/file1.csv'," +
                                "   'format' = 'csv')"
                );

        streamTableEnv
                .executeSql(
                        "CREATE TABLE result_table (" +
                                "  `username` STRING," +
                                "  `count` BIGINT," +
                                "  `window_start_time` TIMESTAMP(3)," +
                                "  `window_end_time` TIMESTAMP(3))" +
                                " WITH ('connector' = 'print')"
                );

        streamTableEnv
                .executeSql(
                        "INSERT INTO result_table" +
                                "  SELECT `username`, COUNT(`username`) AS `count`, " +
                                "  TUMBLE_START(`ts`, INTERVAL '5' SECONDS) AS `window_start_time`," +
                                "  TUMBLE_END(`ts`, INTERVAL '5' SECONDS) AS `window_end_time`" +
                                "  FROM clicks GROUP BY" +
                                "  `username`," +
                                "  TUMBLE(`ts`, INTERVAL '5' SECONDS)"
                );
    }
}
